package com.tmsb.sink;

import com.tmsb.pojo.TaimuMainUserTiXian;
import com.tmsb.utils.ConfUtil;
import com.tmsb.utils.RedisCluster;
import com.tmsb.utils.Util;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.JedisCluster;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;

/**
 * Description: 用taimu_main的user_tixian来获取用户每日提现和费用两个实时指标的sink
 * Version:1.0.0
 *
 * @author wxw
 * @date 2020/7/11
 */
public class UserTiXianSink extends RichSinkFunction<TaimuMainUserTiXian> implements CheckpointedFunction {
    private ListState<BigDecimal> amountValueState;
    private ListState<BigDecimal> feeValueState;
    private ListState<Long> dateState;
    private BigDecimal amountValue;
    private BigDecimal feeValue;
    private Long dateValue;
    private JedisCluster jedisCluster;
    private SimpleDateFormat sdf;

    @Override
    public void invoke(TaimuMainUserTiXian value, Context context) throws Exception {
        //每天0时清空状态
         long eventTime = value.getDate();
         if(eventTime>=dateValue){
             amountValue = new BigDecimal(0);
             feeValue = new BigDecimal(0);
             dateValue = Util.getTomorrowZeroPointTimestamps();
         }else  if(eventTime < Util.getZeroPointTimestamps()){
             //如果事件时间是今天凌晨之前的时间，也就是历史数据，则不进行累加统计
             return;
         }

         //每日用户提现以及费用
         BigDecimal amount = value.getAmount();
         amountValue = amountValue.add(amount);
         BigDecimal fee = value.getFee();
         feeValue =  feeValue.add(fee);

         //向redis发送数据
         String format = sdf.format(new Date(eventTime));
         jedisCluster.hset(ConfUtil.getUserTiXianAmountRedisKey(),format,amountValue+"");
         jedisCluster.hset(ConfUtil.getUserTiXianFeeRedisKey(),format,feeValue+"");

    }

    @Override
    public void close() throws Exception {
        super.close();
        jedisCluster.close();
    }
    //每次请求checkpoint快照时，调用此方法
    @Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        amountValueState.clear();
        feeValueState.clear();
        dateState.clear();
        amountValueState.add(amountValue);
        feeValueState.add(feeValue);
        dateState.add(dateValue);
    }
    //在分布式执行期间创建并行功能实例时，将调用此方法。
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        jedisCluster = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
        sdf = new SimpleDateFormat("yyyyMMdd");
        amountValue = new BigDecimal(0);
        feeValue = new BigDecimal(0);
        dateValue = 0L;
        ListStateDescriptor<BigDecimal> amountStateDes =
                new ListStateDescriptor<>("amountState", TypeInformation.of(new TypeHint<BigDecimal>() {}));
        ListStateDescriptor<BigDecimal> feeStateDes =
                new ListStateDescriptor<>("feeState", TypeInformation.of(new TypeHint<BigDecimal>() {}));
        ListStateDescriptor<Long> dateStateDes =
                new ListStateDescriptor<>("dateSatte", TypeInformation.of(new TypeHint<Long>() {}));
        amountValueState = context.getOperatorStateStore().getListState(amountStateDes);
        feeValueState = context.getOperatorStateStore().getListState(feeStateDes);
        dateState = context.getOperatorStateStore().getListState(dateStateDes);

        if(context.isRestored()){
            Iterator<BigDecimal> iterator = amountValueState.get().iterator();
            if(iterator.hasNext()){
                amountValue = iterator.next();
                feeValue = feeValueState.get().iterator().next();
                dateValue = dateState.get().iterator().next();
            }
        }

    }
}
